#include"TcpConnection.h"
#include"channel.h"
#include"myepoll.h"
#include"Socket.h"
#include<memory>
#include<iostream>
#include"unistd.h"
#include<string.h>
#include"EventLoop.h"
#include<thread>

#include"agreement.h"

#define maxReadChar 1024*20

TcpConnection::TcpConnection(/* args */)
{
}

TcpConnection::TcpConnection(EventLoop *_loop, Socket *_clntSock):loop(_loop),clntSock(_clntSock)
{
    //绑定读
    clntCh = new MyChannel(loop,clntSock->getFd());
    std::function<void()> cb = std::bind(&TcpConnection::readEcho,this,clntSock->getFd());
    clntCh->enableReading();
    clntCh->setReadCallBack(cb);
    //绑定写
    std::function<void()> cd = std::bind(&TcpConnection::writeEcho,this,clntSock->getFd());
    clntCh->setWriteCallBack(cd);
}

TcpConnection::~TcpConnection()
{
    //当客户端断开连接的时候，将socket信息，通道信息都清除
    if (clntSock) {
        delete clntSock;
        clntSock = nullptr;
    }
    if (clntCh) {
        delete clntCh;
        clntCh = nullptr;
    }
}
// 处理客户端的事物 读
void TcpConnection::readEcho(int sockfd)
{
    char buff[maxReadChar] = {0};
    readBuff.BuffetClear();
    while (true) {
        bzero(&buff,sizeof(buff));
        ssize_t readdata = read(sockfd, buff, sizeof(buff));
        if(readdata > 0){
            readBuff.append(buff,readdata);
        }
        else if (readdata == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
        // 已读取完所有数据
            if(dataHandlerCallback){
                dataHandlerCallback(readBuff);
            }
            readBuff.BuffetClear();
            // writeBuff.BuffetClear();
            break;
        } 
        else if(readdata == -1 && errno == EINTR){
            continue;
        }
        else if (readdata == 0) {
            // 客户端断开连接或发生错误
            deleteConnectionCallback(clntSock); 
            break;
        }
    }
    
}
//写数据
void TcpConnection::writeEcho(int sockfd)
{
    //将线上数据提取到本地，减少锁的时间
    std::queue<CloudDrive::Packet> localQueue;
    {
        std::lock_guard<std::mutex> lock(sendMutex);
        std::swap(localQueue,sendQueue);
    }
    //处理队列中的任务
    while (!localQueue.empty())
    {
        const CloudDrive::Packet pkt = localQueue.front();
        sendData(pkt);
        localQueue.pop();
    }
    
    {
        std::lock_guard<std::mutex> lock(sendMutex);
        if(sendQueue.empty()){
            sending_ = false;
            clntCh->enableReading();
        }
    }
    
}

void TcpConnection::setDeleteConnectionCallback(std::function<void(Socket *)> _cb)
{
    deleteConnectionCallback = _cb;
}

void TcpConnection::setDataHandleCallback(std::function<void(Buffer &)> _cb)
{
    dataHandlerCallback = _cb;
}

void TcpConnection::sendData(const CloudDrive::Packet &packet)
{
    std::vector<uint8_t> buffer = CloudDrive::serialize(packet);
    // 发送
    if(clntSock == nullptr){
        std::cout<<"cuowu"<<std::endl;
        return;
    }
    if(clntSock->getFd()==-1 || clntSock->getFd()== 0){
        std::cout<<"关闭连接"<<std::endl;
        return;
    }
    try
    {
        send(clntSock->getFd(), reinterpret_cast<const char*>(buffer.data()), static_cast<int>(buffer.size()), 0);
    }
    catch(const std::exception& e)
    {
        std::cerr << e.what() << '\n';
    }
}

void TcpConnection::enqueueSend(const CloudDrive::Packet &packet)
{
    {
        std::unique_lock<std::mutex> lock(sendMutex);
        sendQueue.push(packet);
    }

    if(!sending_){
        sending_ = true;
        clntCh->enableWriting();
    }
}

int TcpConnection::getFd()
{
    return clntSock->getFd();
}
